package com.intct.flink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-08-08 09:44
 */
public class FlinkSqlCDC {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT, 8081);

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 开启检查点 (FlinkSQL时流程捕获数据，必须要开始检查)
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 创建SQL环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 注册表(临时、永久)
//        tenv.createTemporaryTable();
        tenv.executeSql("CREATE TABLE flink_test03 (\n" +
                "    db_name STRING METADATA FROM 'database_name' VIRTUAL\n" +
                "    ,table_name STRING METADATA  FROM 'table_name' VIRTUAL\n" +
                "    ,operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL\n" +
                "    ,id int\n" +
                "    ,name string\n" +
                "    ,PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'mysql-cdc'\n" +
                "    ,'hostname' = 'cdh-node'\n" +
                "    ,'port' = '13306'\n" +
                "    ,'username' = 'root'\n" +
                "    ,'password' = 'Test_090110'\n" +
                "    ,'database-name' = 'm1'\n" +
                "    ,'table-name' = 'test03'\n" +
                "    ,'scan.incremental.snapshot.chunk.key-column' = 'id'\n" +
                ")");

        // 输出
        tenv.executeSql("insert into kafak_sink select * from flink_test03").print();
        env.execute();
    }
}
